package com.aiwiown.snackmq.storage.timer;

import com.aiwiown.snackmq.storage.util.BufferCleaner;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.Buffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 代表定时消息存储的一个段文件。
 * 每个段文件包含多个固定大小的 TimerLogEntry。
 */
@Slf4j
@Getter
public class TimerLogSegment implements Closeable {

    public static final int TIMER_LOG_ENTRY_SIZE = 16; // long (deliveryTimestamp) + long (commitLogOffset)

    private final File file;
    private final RandomAccessFile randomAccessFile;
    private final FileChannel fileChannel;
    private final MappedByteBuffer mappedBuffer;
    private final long baseOffset;
    private final long segmentSize;
    private final AtomicInteger writePosition;
    private final AtomicBoolean closed = new AtomicBoolean(false);

    public TimerLogSegment(String path, long baseOffset, long segmentSize) throws IOException {
        this.baseOffset = baseOffset;
        this.segmentSize = segmentSize;
        this.file = new File(path);

        boolean fileExisted = file.exists();
        this.randomAccessFile = new RandomAccessFile(file, "rw");
        this.fileChannel = randomAccessFile.getChannel();
        this.mappedBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, segmentSize);

        if (fileExisted) {
            this.writePosition = new AtomicInteger(recover());
        } else {
            this.writePosition = new AtomicInteger(0);
        }
    }

    private int recover() {
        int position = 0;
        // 扫描文件，找到第一个为空的条目，即为下一个写入点
        while (position + TIMER_LOG_ENTRY_SIZE <= this.mappedBuffer.limit()) {
            long deliveryTimestamp = this.mappedBuffer.getLong(position);
            if (deliveryTimestamp == 0) {
                break; // 找到文件末尾的空条目
            }
            position += TIMER_LOG_ENTRY_SIZE;
        }
        log.debug("Recovered timer log segment {}. Write position is {}", file.getName(), position);
        return position;
    }

    /**
     * 追加一个定时消息条目到段文件中。
     *
     * @param deliveryTimestamp 消息的投递时间戳
     * @param commitLogOffset   消息在主 CommitLog 中的物理偏移量
     * @return 如果成功追加返回 true，如果段已满则返回 false
     */
    public boolean appendEntry(long deliveryTimestamp, long commitLogOffset) {
        if (isFull()) {
            return false;
        }
        int currentPosition = writePosition.getAndAdd(TIMER_LOG_ENTRY_SIZE);

        // 【最终修复】: 强制类型转换为 Buffer 以兼容 JDK 8。
        // JDK 9+ 的 MappedByteBuffer.position(int) 返回 MappedByteBuffer，而 JDK 8 返回 Buffer。
        // 这种不匹配会导致在 JDK 8 环境下运行时出现 NoSuchMethodError。
        ((Buffer) mappedBuffer).position(currentPosition);

        mappedBuffer.putLong(deliveryTimestamp);
        mappedBuffer.putLong(commitLogOffset);
        return true;
    }

    /**
     * 从指定的物理位置读取所有后续的条目。
     *
     * @param physicalPosition 物理读取起始点
     * @return 读取到的条目列表
     */
    public List<TimerMessageStore.TimerLogEntry> getEntries(int physicalPosition) {
        List<TimerMessageStore.TimerLogEntry> entries = new ArrayList<>();
        int position = physicalPosition;
        // 使用 duplicate() 创建一个只读的、线程安全的视图
        MappedByteBuffer readOnlyBuffer = (MappedByteBuffer) this.mappedBuffer.asReadOnlyBuffer();

        while (position + TIMER_LOG_ENTRY_SIZE <= writePosition.get()) {
            long deliveryTimestamp = readOnlyBuffer.getLong(position);
            long commitLogOffset = readOnlyBuffer.getLong(position + 8);

            if (deliveryTimestamp == 0) {
                // 遇到空条目，说明是槽的末尾
                break;
            }
            // 【最终修复】: 传入第三个参数 `position`，以匹配新的构造函数签名。
            entries.add(new TimerMessageStore.TimerLogEntry(deliveryTimestamp, commitLogOffset, position));
            position += TIMER_LOG_ENTRY_SIZE;
        }
        return entries;
    }

    /**
     * 清除（置零）指定位置开始的N个条目。
     *
     * @param physicalPosition 物理起始点
     * @param count            要清除的条目数量
     */
    public void clearEntries(int physicalPosition, int count) {
        if (count == 0) return;
        int position = physicalPosition;
        for (int i = 0; i < count; i++) {
            // 【最终修复】: 同样进行强制类型转换
            ((Buffer) mappedBuffer).position(position);
            mappedBuffer.putLong(0L); // 将投递时间置零，标记为已处理
            mappedBuffer.putLong(0L); // 将偏移量置零
            position += TIMER_LOG_ENTRY_SIZE;
        }
    }

    /**
     * 检查段是否已满。
     *
     * @return 如果已满返回 true
     */
    public boolean isFull() {
        return writePosition.get() + TIMER_LOG_ENTRY_SIZE > segmentSize;
    }

    /**
     * 将缓冲区内容强制刷入磁盘。
     */
    public void flush() {
        if (mappedBuffer != null) {
            mappedBuffer.force();
        }
    }

    @Override
    public void close() throws IOException {
        if (closed.compareAndSet(false, true)) {
            log.debug("Closing timer log segment: {}", file.getName());
            flush();
            BufferCleaner.clean(mappedBuffer);
            if (fileChannel != null && fileChannel.isOpen()) {
                fileChannel.close();
            }
            if (randomAccessFile != null) {
                randomAccessFile.close();
            }
        }
    }
}